package com.xjwn.datasync;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.toolkit.Db;
import com.xjwn.datasync.mapper.DatasyncLogMapper;
import com.xjwn.datasync.mapper.DatasyncMapper;
import com.xjwn.datasync.pojo.Datasync;
import com.xjwn.datasync.pojo.DatasyncBi;
import com.xjwn.datasync.pojo.DatasyncDb;
import com.xjwn.datasync.pojo.DatasyncLog;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.lang.Nullable;
import org.springframework.web.bind.annotation.RestController;

import javax.sql.rowset.CachedRowSet;
import javax.sql.rowset.RowSetProvider;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.Date;

@Slf4j
@RestController("insert")
public class Task1 {

    @Autowired
    private DatasyncLogMapper datasyncLogMapper;
    @Autowired
    private DatasyncMapper datasyncMapper;

    /**
     * SqlServer
     *
     * @param params    目标表名
     * @param sdburl    源数据库连接
     * @param tdburl    目标数据库连接
     * @param remark    任务名称
     * @param begintime 存储过程参数1：开始时间
     * @param endtime   存储过程参数2：结束时间
     * @param proc      存储过程名称
     * @param pkname    主键
     */
    public void SqlServer(String params, String sdburl, String tdburl, String remark, String begintime, String endtime, String proc, String pkname, String yydm, String yymc, String isonce) {

        System.out.println("SqlServer参数：" + params + "----" + sdburl + "======" + tdburl);
        String uuid = UUID.randomUUID().toString();
        String sqltext = "";
        try {

            //region 源数据库连接信息
            DatasyncDb sdb = Db.lambdaQuery(DatasyncDb.class)
                    .eq(DatasyncDb::getName, sdburl)
                    .one();
            String s_url = "";
            String s_user = sdb.getUsername();
            String s_password = sdb.getPassword();
            String s_dbname = sdb.getDbname();
            if ("Oracle".equalsIgnoreCase(sdb.getType())) {
                s_url = "jdbc:oracle:thin:@//" + sdb.getUrl();
            } else if ("SqlServer".equalsIgnoreCase(sdb.getType())) {
                s_url = String.format("jdbc:sqlserver://%s;databaseName=%s;encrypt=false;trustServerCertificate=True;loginTimeout=120;",
                        sdb.getUrl(), s_dbname);
            } else {
                throw new IllegalArgumentException("不支持的数据库类型");
            }
            //endregion

            //region 源数据库连接信息
            DatasyncDb tdb = Db.lambdaQuery(DatasyncDb.class)
                    .eq(DatasyncDb::getName, tdburl)
                    .one();
            String t_url = "";
            String t_user = tdb.getUsername();
            String t_password = tdb.getPassword();
            String t_dbname = tdb.getDbname();
            if ("Oracle".equalsIgnoreCase(tdb.getType())) {
                t_url = "jdbc:oracle:thin:@//" + tdb.getUrl();
            } else if ("SqlServer".equalsIgnoreCase(tdb.getType())) {
                t_url = String.format("jdbc:sqlserver://%s;databaseName=%s;encrypt=false;trustServerCertificate=True;loginTimeout=120;",
                        tdb.getUrl(), t_dbname);
            } else {
                throw new IllegalArgumentException("不支持的数据库类型");
            }
            //endregion


            // 建立数据库连接
            Connection scon = DriverManager.getConnection(s_url, s_user, s_password);
            System.out.println("源数据库：" + sdburl);
            System.out.println("源数据库连接成功");

            // 建立数据库连接
            Connection tcon = DriverManager.getConnection(t_url, t_user, t_password);
            System.out.println("目标数据库：" + tdburl);
            System.out.println("目标数据库连接成功");

            // 执行查询
            Statement stmt = scon.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
//            Statement stmttime = conn.createStatement();
//            ResultSet rs2 = stmt.executeQuery("select top 1 * from datasync where params='"+params+"' and isonce=0 order by time desc");
//            QueryWrapper<Datasync> wrapper = new QueryWrapper<>();
//            wrapper.orderByDesc("time")
//                    .allEq(Map.of("params", params, "isonce", 0));
//            Datasync datasync = datasyncMapper.selectOne(wrapper);
//            ResultSet rs = stmt.executeQuery("SELECT * FROM schedule_setting");
            String kssj;
            String jssj;

            // 获取当前时间
            LocalDateTime now = LocalDateTime.now();
            // 定义格式化模式（注意大小写）
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            // 格式化时间
            String formattedTime = now.format(formatter);
            System.out.println("时间：" + formattedTime);
            // 示例输出：20231005143045（假设当前时间是 2023-10-05 14:30:45）

            //判断是不是执行单次
            if (isonce == "false") {
                //任务定时执行
                Datasync datasync = datasyncMapper.selecttop1(params);
                if (datasync != null) {
                    kssj = datasync.getJssj();
                    jssj = formattedTime;
                } else {
                    kssj = begintime;
                    jssj = endtime;
                }
            } else {
                //单次执行
                kssj = begintime;
                jssj = endtime;
            }
            sqltext = "EXEC " + proc +
                    " @yljgdm='" + yydm + "'," +
                    "@yljgmc='" + yymc + "'," +
                    "@kssj='" + kssj + "'," +
                    "@jssj='" + jssj + "'";
            System.out.println(sqltext);
            System.out.println("11111111111111111111111111111111111111111111111111111");
            stmt.setQueryTimeout(300);
//            ResultSet rs = stmt.executeQuery(sqltext);

            ResultSet rs = stmt.executeQuery("EXEC usp_wngwpt_YP_GYSXXK @yljgdm='12345677',@yljgmc='测试机构',@kssj='2025-08-08 00:00:00',@jssj='2025-08-08 13:14:39'");
            CachedRowSet crs = RowSetProvider.newFactory().createCachedRowSet();
            crs.populate(rs);
//            while (rs.next()) {
//                System.out.println(rs.getString("sdb_url")); // 替换为实际列名
//            }
            if (!crs.next()) {
                log.warn("结果集为空：{}", sqltext);
            } else {
                log.info("存储执行成功：{}", sqltext);
                crs.beforeFirst();
                // 3. 生成INSERT语句
                List<List<String>> insertStatements = generateInsertStatements(crs, params, pkname, "SqlServer");

                // 4. 输出或执行生成的SQL
//            System.out.println("📄 生成的INSERT语句：");
//            insertStatements.forEach(System.out::println);
//            System.out.println("生成的INSERT语句成功");
//            System.out.println("行数：" + insertStatements.stream().count());


                // 5. （可选）执行生成的SQL
                DatasyncBi datasyncBi = executeBatch(scon, tcon, insertStatements, params, pkname, kssj, jssj, uuid);
                log.info("执行成功：保存任务执行信息开始");
                Datasync datasync1 = new Datasync(params,kssj,jssj,formattedTime,datasyncBi.getCount(),datasyncBi.getSuccess(),datasyncBi.getError(),datasyncBi.getUpdate(),isonce=="false"?0:1,uuid,sqltext.replace("'","''"));
                datasync1.insert();
                log.info("执行成功：保存任务执行信息结束");
//            System.out.println("数据库连接成功");
                // 关闭资源

            }
            rs.close();
            stmt.close();
            scon.close();
            tcon.close();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }


    /**
     * Oracle
     *
     * @param params    目标表名
     * @param sdburl    源数据库连接
     * @param tdburl    目标数据库连接
     * @param remark    任务名称
     * @param begintime 存储过程参数1：开始时间
     * @param endtime   存储过程参数2：结束时间
     * @param proc      存储过程名称
     */
    public void Oracle(String params, String sdburl, String tdburl, String remark, String begintime, String endtime, String proc, String pkname, String yydm, String yymc, String isonce) {
//        log.info("taskByParams执行时间:{}", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
//        log.info("taskByParams执行有参示例任务：{}",params+"曹尼玛");
        System.out.println("SqlServer参数：" + params + "----" + sdburl + "======" + tdburl);
        String uuid = UUID.randomUUID().toString();
        String sqltext = "";
//        String sdburl="jdbc:sqlserver://192.168.4.137:1433;databaseName=datasync;username=sa;password=Sq_123456;encrypt=false;trustServerCertificate=True;loginTimeout=30; ";
        try {

            //region 源数据库连接信息
            DatasyncDb sdb = Db.lambdaQuery(DatasyncDb.class)
                    .eq(DatasyncDb::getName, sdburl)
                    .one();
            String s_url = "";
            String s_user = sdb.getUsername();
            String s_password = sdb.getPassword();
            String s_dbname = sdb.getDbname();
            if ("Oracle".equalsIgnoreCase(sdb.getType())) {
                s_url = "jdbc:oracle:thin:@//" + sdb.getUrl();
            } else if ("SqlServer".equalsIgnoreCase(sdb.getType())) {
                s_url = String.format("jdbc:sqlserver://%s;databaseName=%s;encrypt=false;trustServerCertificate=True;loginTimeout=120;",
                        sdb.getUrl(), s_dbname);
            } else {
                throw new IllegalArgumentException("不支持的数据库类型");
            }
            //endregion

            //region 源数据库连接信息
            DatasyncDb tdb = Db.lambdaQuery(DatasyncDb.class)
                    .eq(DatasyncDb::getName, tdburl)
                    .one();
            String t_url = "";
            String t_user = tdb.getUsername();
            String t_password = tdb.getPassword();
            String t_dbname = tdb.getDbname();
            if ("Oracle".equalsIgnoreCase(tdb.getType())) {
                t_url = "jdbc:oracle:thin:@//" + tdb.getUrl();
            } else if ("SqlServer".equalsIgnoreCase(tdb.getType())) {
                t_url = String.format("jdbc:sqlserver://%s;databaseName=%s;encrypt=false;trustServerCertificate=True;loginTimeout=120;",
                        tdb.getUrl(), t_dbname);
            } else {
                throw new IllegalArgumentException("不支持的数据库类型");
            }
            //endregion


            // 建立数据库连接
            Connection conn = DriverManager.getConnection(s_url, s_user, s_password);
            System.out.println("源数据库：" + sdburl);
            System.out.println("源数据库连接成功");

            // 建立数据库连接
            Connection conn2 = DriverManager.getConnection(t_url, t_user, t_password);
            System.out.println("目标数据库：" + tdburl);
            System.out.println("目标数据库连接成功");

            // 执行查询
            Statement stmt = conn.createStatement();
//            Statement stmttime = conn.createStatement();
//            ResultSet rs2 = stmt.executeQuery("select top 1 * from datasync where params='"+params+"' and isonce=0 order by time desc");
//            QueryWrapper<Datasync> wrapper = new QueryWrapper<>();
//            wrapper.orderByDesc("time")
//                    .allEq(Map.of("params", params, "isonce", 1));
//            log.info("任务{}查询是否有上传记录",remark);
//            Datasync datasync = datasyncMapper.selectOne(wrapper);
//            ResultSet rs = stmt.executeQuery("SELECT * FROM schedule_setting");
            String kssj;
            String jssj;

            // 获取当前时间
            LocalDateTime now = LocalDateTime.now();
            // 定义格式化模式（注意大小写）
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            // 格式化时间
            String formattedTime = now.format(formatter);
            System.out.println("时间：" + formattedTime);
            // 示例输出：20231005143045（假设当前时间是 2023-10-05 14:30:45）

            //判断是不是执行单次
            if (isonce == "false") {
                Datasync datasync = datasyncMapper.selecttop1(params);
                //任务定时执行
                if (datasync != null) {
                    kssj = datasync.getJssj();
                    jssj = formattedTime;
                } else {
                    kssj = begintime;
                    jssj = endtime;
                }
            } else {
                //单次执行
                kssj = begintime;
                jssj = endtime;
            }
            sqltext = "EXEC " + proc + " " +
                    "@yljgdm='" + yydm + "'," +
                    "@yljgmc='" + yymc + "'," +
                    "@kssj='" + kssj + "'," +
                    "@jssj='" + jssj + "'";
            System.out.println(sqltext);
            System.out.println("11111111111111111111111111111111111111111111111111111");
            stmt.setQueryTimeout(600);

            ResultSet rs = stmt.executeQuery(sqltext);
            CachedRowSet crs = RowSetProvider.newFactory().createCachedRowSet();
            crs.populate(rs);
//            int i=0;
//            while (crs.next()) { // 实际从第二行开始循环
//                i++;
//                log.info("内容：{}",i);
//            }
//             i=0;
//            crs.beforeFirst();
//            while (crs.next()) { // 实际从第二行开始循环
//                i++;
//                log.info("内容：{}",i);
//            }
//            crs.absolute(1);
            if (!crs.next()) {
                log.warn("结果集为空：{}", sqltext);
            } else {
                log.info("存储执行成功：{}", sqltext);
//            while (rs.next()) {
//                System.out.println(rs.getString("sdb_url")); // 替换为实际列名
//            }
                crs.beforeFirst();
                // 3. 生成INSERT语句

                List<List<String>> insertStatements = generateInsertStatements(crs, params, pkname, "Oracle");

                // 4. 输出或执行生成的SQL
                System.out.println("📄 生成的INSERT语句：");
                insertStatements.forEach(System.out::println);
                System.out.println("生成的INSERT语句成功");
                System.out.println("行数：" + insertStatements.stream().count());

//                int count = insertStatements.size());;
//                int success;
//                int error;
//                int update;
                // 5. （可选）执行生成的SQL
                DatasyncBi datasyncBi = executeBatch(conn, conn2, insertStatements, params, pkname, kssj, jssj, uuid);
                log.info("任务：{},总量：{}；插入：{}；更新：{}；失败：{}",params,datasyncBi.getCount(),datasyncBi.getSuccess(),datasyncBi.getUpdate(),datasyncBi.getError());
                log.info("执行成功：保存任务执行信息开始{}至{}",kssj,jssj);
                Datasync datasync1 = new Datasync(params,kssj,jssj,formattedTime,datasyncBi.getCount(),datasyncBi.getSuccess(),datasyncBi.getError(),datasyncBi.getUpdate(),isonce=="false"?0:1,uuid,sqltext.replace("'","''"));
                datasync1.insert();
                log.info("执行成功：保存任务执行信息结束");


//                String sqlog = "insert into datasync(params,kssj,jssj) values ('" + params + "','" + begintime + "','" + endtime + "')";
//                PreparedStatement pstmt3 = null;
//                pstmt3 = conn.prepareStatement(sqlog);
//                pstmt3.executeUpdate();
//            System.out.println("数据库连接成功");
                // 关闭资源
            }
            rs.close();
            stmt.close();
            conn.close();
            conn2.close();
            crs.close();
        } catch (Exception e) {
            System.out.println(e.toString());
            log.error("任务'{}'执行失败，存储：{}，失败原因：{}", params, sqltext, e.toString());
        }
    }

    public void test(String params) {
        log.info("test执行时间:{}", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        log.info("test执行有参示例任务：{}", params);
    }


    /**
     * 将ResultSet转换为INSERT语句列表
     *
     * @param rs   需要处理的ResultSet
     * @param type SqlServer/Oracle
     * @return INSERT语句列表
     * @throws SQLException 当处理结果集时发生错误
     */
    public static List<List<String>> generateInsertStatements(CachedRowSet rs, String params, String pkname, String type) throws SQLException {
        System.out.println("开始换为INSERT语句：");
        List<List<String>> statements = new ArrayList<>();
        ResultSetMetaData metaData = rs.getMetaData();
        System.out.println("获取列的数量");
        int columnCount = metaData.getColumnCount();

        // 构建列名部分
        System.out.println("构建列名部分");
        StringBuilder columns = new StringBuilder();
        for (int i = 1; i <= columnCount; i++) {
            columns.append(metaData.getColumnName(i));
//                    .append("(")
//                    .append(")");
            if (i < columnCount) columns.append(", ");
            System.out.println("构建的列名："+columns);
        }

        // 构建值模板
        StringBuilder valuesTemplate = new StringBuilder();
        for (int i = 1; i <= columnCount; i++) {
            valuesTemplate.append("?");
            if (i < columnCount) valuesTemplate.append(", ");
        }
        System.out.println("aaa");
        String insertTemplate = String.format(
                "INSERT INTO %s (%s) VALUES (%s)",
                params,  //表名
//                metaData.getTableName(1),  // 获取表名
                columns.toString(),
                valuesTemplate.toString()
        );

        // 处理每一行数据
        int rowCount = 0;
        System.out.println("bbb");
        rs.beforeFirst();
        while (rs.next()) {
            rowCount++;
            StringBuilder values = new StringBuilder();
            System.out.println("处理每一行数据-开始");
            for (int i = 1; i <= columnCount; i++) {
                Object value = rs.getObject(i);
                values.append(formatValue(value, metaData, i, type))
                        .append(",");
                System.out.println(values.toString());
            }
            System.out.println("处理每一行数据-结束");
            // 移除最后一个逗号
            System.out.println("ccc");
            if (values.length() > 0) values.setLength(values.length() - 1);
            String fullSql = insertTemplate.replace(valuesTemplate.toString(), values.toString());
            System.out.println(fullSql);
            List<String> pknlist = Arrays.asList(pkname.split(","));
//            List<String> pknlist = Arrays.asList(rs.getString(pkname).split(","));
            String string="";
            for (int i=0;i<pknlist.size();i++){
                string=string+rs.getString(pknlist.get(i))+",";
            }
            if (string.length() > 0) values.setLength(string.length() - 1);
//            statements.add(new ArrayList<>(Arrays.asList(rs.getString(pkname), fullSql)));
            statements.add(new ArrayList<>(Arrays.asList(string, fullSql)));
        }
        System.out.println("换为INSERT语句结束：");
        log.info("params：{}，数量：{}",params,rowCount);
//        System.out.printf("✅ 生成%d条INSERT语句%n", rowCount);
        rs.close();
        return statements;
    }

    /**
     * 格式化单个值（处理NULL和特殊类型）
     *
     * @param value       原始值
     * @param metaData    元数据
     * @param columnIndex 列索引
     * @return 格式化后的值字符串
     * @throws SQLException 当处理值时发生错误
     */
    private static String formatValue(Object value, ResultSetMetaData metaData, int columnIndex, String type)
            throws SQLException {
        if (value == null) {
            return "NULL";
        }
        if (value.toString().length() > 8) {
            String ss = value.toString();
            if (ss.length() > 8 && ss.charAt(4) == '-' && ss.charAt(7) == '-' && "Oracle".equals(type)) {
                return "to_date('" + ss + "','yyyy-mm-dd hh24:mi:ss')";
            }
            else if(ss.length() < 11 && ss.charAt(4) == '-' && ss.charAt(7) == '-' && "Oracle".equals(type)){
                return "to_date('" + ss+ "','yyyy-mm-dd')";
            }
        }


        int sqlType = metaData.getColumnType(columnIndex);
        switch (sqlType) {
            case Types.INTEGER:
            case Types.BIGINT:
            case Types.SMALLINT:
            case Types.TINYINT:
                return String.valueOf(value);

            case Types.VARCHAR:
            case Types.NVARCHAR:
//            case Types.TEXT:
                String strValue = value.toString();
                return "'" + escapeSingleQuotes(strValue) + "'";

            case Types.BOOLEAN:
                return value.equals(true) ? "TRUE" : "FALSE";

            case Types.TIMESTAMP:
            case Types.DATE:
            case Types.TIME:
                Timestamp ts = (Timestamp) value;
                return "'" + ts.toString() + "'";

            default:
                return "'" + value.toString() + "'";
        }
    }

    /**
     * 转义单引号（防止SQL注入）
     *
     * @param str 输入字符串
     * @return 转义后的字符串
     */
    private static String escapeSingleQuotes(String str) {
        if (str == null) return "";
        return str.replaceAll("'", "''");
    }

    /**
     * 执行生成的批量INSERT语句（可选）
     *
     * @param conn       数据库连接
     * @param statements INSERT语句列表
     * @throws SQLException 当执行SQL时发生错误
     */
    private static DatasyncBi executeBatch(Connection conn, Connection conn2, List<List<String>> statements,
                                           String params, String pkname, String begintime, String endtime, String uuid) throws SQLException {
        System.out.println("========执行sql开始========");
// 获取当前时间
        LocalDateTime now = LocalDateTime.now();
        // 定义格式化模式（注意大小写）
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        // 格式化时间
        String formattedTime = now.format(formatter);
        // 统计变量
        int sum = 0;
        int success = 0;
        int error = 0;
        int update = 0;

        // 提取主键列表和SQL列表（提前处理，避免循环内重复判断）
        List<String> pknamelist = new ArrayList<>();
        List<String> sqllist = new ArrayList<>();
        for (List<String> sublist : statements) {
            if (sublist.size() > 1) {
                pknamelist.add(sublist.get(0));
                sqllist.add(sublist.get(1));
            }
        }
        sum = sqllist.size();
        if (sum == 0) {
            System.out.println("没有需要执行的SQL");
            return new DatasyncBi(sum, success, update, error);
//            return  new  Datasync(params,begintime,endtime,formattedTime,sum,success,error,update)
        }

        // 解析主键名（提前解析，避免循环内重复split）
        List<String> pknlist = Arrays.asList(pkname.split(","));

        // 预编译SQL（复用PreparedStatement，减少对象创建开销）
        PreparedStatement queryStmt = null;   // 用于查询数据是否存在
        PreparedStatement deleteStmt = null;  // 用于删除数据
        PreparedStatement insertStmt = null;  // 用于插入数据

        try {
            // 优化：关闭自动提交，每执行N条手动提交一次（平衡性能和事务安全性）
            conn2.setAutoCommit(false);
            int commitBatchSize = 100;  // 每100条提交一次事务
            int currentCommitCount = 0;

            // 循环处理每条SQL
            for (int i = 0; i < sqllist.size(); i++) {
                String sql = sqllist.get(i);
                String pkValue = pknamelist.get(i);
                System.out.println("目标数据库执行语句: " + sql);

                DatasyncLog datasyncLog = new DatasyncLog();
                try {
                    // 1. 构建查询条件（复用方法，减少重复代码）
                    String wherestr = buildWhereClause(pknlist, pkValue);
                    String querySql = "select 1 from " + params + wherestr;  // 只查存在性，不查全部字段

                    // 2. 查询数据是否存在（复用PreparedStatement）
                    queryStmt = conn2.prepareStatement(querySql);
                    boolean exists;
                    try (ResultSet rs = queryStmt.executeQuery()) {
                        exists = rs.next();  // 只需要判断是否有记录
                    }

                    // 3. 根据存在性执行操作
                    if (exists) {
                        // 已存在：先删除再插入（更新逻辑）
                        String deleteSql = "delete from " + params + wherestr;
                        deleteStmt = conn2.prepareStatement(deleteSql);
                        deleteStmt.executeUpdate();

                        insertStmt = conn2.prepareStatement(sql);
                        insertStmt.executeUpdate();
                        update++;
                    } else {
                        // 不存在：直接插入
                        insertStmt = conn2.prepareStatement(sql);
                        insertStmt.executeUpdate();
                        success++;
                    }

                    // 4. 记录成功日志
                    recordLog(datasyncLog, uuid, pkValue, sql, params, "0", "");
                    currentCommitCount++;

                    // 5. 达到提交批次，手动提交事务
                    if (currentCommitCount % commitBatchSize == 0) {
                        conn2.commit();
                        System.out.println("✅ 已提交" + currentCommitCount + "条数据");
                    }

                } catch (SQLException e) {
                    error++;
                    // 记录错误日志
                    recordLog(datasyncLog, uuid, pkValue, sql, params, "1", e.toString().replace("'", "''"));
                    // 出错时回滚当前批次，避免部分提交
//                    conn2.rollback();
                    System.err.println("❌ 第" + (i + 1) + "条执行失败: " + e.getMessage());
                } finally {
                    // 及时关闭Statement，避免资源泄露（但保留连接）
                    closeStatement(insertStmt);
                    closeStatement(deleteStmt);
                    closeStatement(queryStmt);
                }
            }

            // 提交剩余未提交的数据
            if (currentCommitCount > 0) {
                conn2.commit();
                System.out.println("✅ 最终提交，共处理" + currentCommitCount + "条数据");
            }

        } finally {
            // 确保所有资源关闭
            closeStatement(insertStmt);
            closeStatement(deleteStmt);
            closeStatement(queryStmt);
            closeConnection(conn);
            closeConnection(conn2);
        }

        log.info("任务{}，总量{}，插入{}，更新{}，错误{}", uuid, sum, success, update, error);
        return new DatasyncBi(sum, success, error, update);
    }

    // 工具方法：构建WHERE条件
    private static String buildWhereClause(List<String> pknlist, String pkValue) {
        List<String> pklist = Arrays.asList(pkValue.split(","));
        StringBuilder wherestr = new StringBuilder(" where ");
        for (int i = 0; i < pknlist.size(); i++) {
            if (i > 0) {
                wherestr.append(" and ");
            }
            wherestr.append(pknlist.get(i)).append("='").append(pklist.get(i)).append("'");
        }
        return wherestr.toString();
    }

    // 工具方法：记录日志
    private static void recordLog(DatasyncLog log, String uuid, String pkname, String sql,
                                  String params, String status, String logMsg) {
        LocalDateTime currentDateTime = LocalDateTime.now();
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        log.setUuid(uuid);
        log.setTime(currentDateTime.format(formatter));
        log.setLog(logMsg);
        log.setPkname(pkname);
        log.setSql(sql);
        log.setParams(params);
        log.setStatus(status);
        log.insert();
    }

    // 工具方法：关闭Statement（避免重复try-catch）
    private static void closeStatement(Statement stmt) {
        if (stmt != null) {
            try {
                stmt.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }

    // 工具方法：关闭Connection
    private static void closeConnection(Connection conn) {
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 统一处理SQL异常
     *
     * @param e 异常对象
     */
    private static void handleSQLException(SQLException e) {
        System.err.println("❌ SQL操作失败");
        System.err.println("错误代码: " + e.getErrorCode());
        System.err.println("错误消息: " + e.getMessage());
        System.err.println("SQL状态: " + e.getSQLState());
        e.printStackTrace();
    }

}